package streaming.day02.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.Future;

/**
 * kafka客户端api - 生产者
 * Created 2018/4/22
 */
public class ProducerApi {

    public static void main(String[] args) throws  Exception{

        // 封装配置参数
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "kk-01:9092,kk-02:9092,kk-03:9092"); // kafka的brokers列表
        // key 和 value的序列化方式
        props.setProperty("key.serializer", StringSerializer.class.getName());
        props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 发送数据的时候做应答
        // [all, -1, 0, 1]
        // 0 leader不做任何应答
        // 1 leader会给producer做出应答
        // all, -1 follower -> leader -> producer
        // props.setProperty("acks", "1");

        // 自定义分区器
        // props.setProperty("partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");

        // 创建一个生产者的客户端实例
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);

        int i = 1;
        while (i < 1000) {

            int partitionNumber = i % 3;

            /**
             * partition: 如果指定了分区编号，则数据肯定写入到该分区下
             * key: 如果没有指定分区，这条会根据key hash决定改条数据去往那个分区 UUID
             * value：数据本省
             */
            // 封装一条消息
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("helloTopic", partitionNumber, "", ""+i);

            // 发送数据
            kafkaProducer.send(record);

            Thread.sleep(1 * 1000);
            i ++;
        }


        kafkaProducer.close();
        System.out.println("end");
    }

}
